package com.chis.spark;

import com.alibaba.fastjson.JSON;
import com.chis.jm.bean.jxc.trace.TraceBean;
import com.chis.jm.bean.ll.MsgBean;
import com.chis.jmdataspark.utils.DecodeingKafka;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.serialization.StringDeserializer;

import java.util.Arrays;
import java.util.Properties;

/**
 * 将kafka 重新发送
 * @author wlj
 * @version 2019年03月26日
 * @Description
 * @Company zwx
 */
public class LlNew {


    public static void main(String[] args) throws Exception{
        String topicName = "TOPIC_LL_WARN";
        Integer part = Integer.valueOf(args[0]);
        Long offset = Long.valueOf(args[1]);
        Properties props = new Properties();
        //用于初始化时建立链接到kafka集群
//        props.put("bootstrap.servers", "10.88.66.110:9093,10.88.66.111:9093,10.88.66.125:9093");
        props.put("bootstrap.servers", "10.41.4.177:7113,10.41.4.178:7113,10.41.4.179:7113,10.41.4.180:7113,10.41.4.181:7113,10.41.4.182:7113,10.41.4.183:7113,10.41.4.184:7113");
//        props.put("bootstrap.servers", "192.168.150.90:9092,192.168.150.91:9092,192.168.150.92:9092");
        //kafka使用消费者分组的概念来允许多个消费者共同消费和处理同一个topic中的消息。分组中消费者成员是动态维护的，如果一个消费者处理失败了，那么之前分配给它的partition将被重新分配给分组中其他消费者；同样，如果分组中加入了新的消费者，也将触发整个partition的重新分配，每个消费者将尽可能的分配到相同数目的partition，以达到新的均衡状态；
        props.put("group.id", "testll");
        //用于配置是否自动的提交消费进度
        props.put("enable.auto.commit", "true");
        //用于配置自动提交消费进度的时间
        props.put("auto.commit.interval.ms", "1000");
        //会话超时时长，客户端需要周期性的发送“心跳”到broker，这样broker端就可以判断消费者的状态，如果消费在会话周期时长内未发送心跳，那么该消费者将被判定为dead，那么它之前所消费的partition将会被重新的分配给其他存活的消费者
        props.put("session.timeout.ms", "30000");
        props.put("key.deserializer", StringDeserializer.class);
        props.put("value.deserializer", DecodeingKafka.class);
        props.put("max.poll.records", 100);

//        ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext(new String[] { "consumer_sparkdeal.xml" });
//        IKafkaMsgService kafkaservice = (IKafkaMsgService) context.getBean("kafkaMsgService");
//        if(kafkaservice == null){
//            throw new RuntimeException("获取kafka服务失败");
//        }

//        ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext(new String[]{"consumer_sparkdeal.xml"});
//        context.start();

        //安全认证
        KafkaConsumer<String, Object> consumer = new KafkaConsumer<>(props);
        //订阅主题列表topic
//        String topicName = "TOPIC_TRACE";
//        //用于分配topic和partition

        ConsumerRecords<String, Object> records = null;

        consumer.assign(Arrays.asList(new TopicPartition(topicName, part)));
        //不改变当前offset，指定从这个topic和partition的开始位置获取。
//        consumer.seekToBeginning(Arrays.asList(new TopicPartition(topicName, 1)));
        consumer.seek(new TopicPartition(topicName, part), offset);

        while (true){
            records = consumer.poll(1000);
            for (ConsumerRecord<String, Object> record : records) {
                System.out.println("=====================================");
                System.out.printf("partition = %d, offset = %d, key = %s, value = %s", record.partition(), record.offset(), record.key(), record.value() + "\n");
                MsgBean jxcSynBean = (MsgBean) record.value();
                String str = JSON.toJSONString(jxcSynBean);
                System.out.println(str);

            }
            if(records.isEmpty()){
                break;
            }
        }





    }
}
